package edu.xhu.behavior.listener;

import com.alibaba.fastjson2.JSON;
import edu.xhu.api.video.VideoCrudClient;
import edu.xhu.behavior.mapper.BiliUserDanmuMapper;
import edu.xhu.common.exception.CustomException;
import edu.xhu.model.behavior.pojos.BiliUserDanmu;
import edu.xhu.model.common.constants.kafka.group.DanmuConsumerGroup;
import edu.xhu.model.common.constants.kafka.topic.DanmuTopic;
import edu.xhu.model.common.dtos.ResponseResult;
import edu.xhu.model.common.enums.AppHttpCodeEnum;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.interceptor.TransactionAspectSupport;

import java.util.ArrayList;
import java.util.List;

/**
 * 弹幕的listener
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class DanmuSaveListener {

    private final BiliUserDanmuMapper biliUserDanmuMapper;

    private final VideoCrudClient videoCrudClient;


    //批处理监听,异步批处理,减少打db次数
    @KafkaListener(topics = DanmuTopic.BILL_Danmu_TOPIC, groupId = DanmuConsumerGroup.BILI_DANMU_GROUP)
    @Transactional
    public void detail_save(List<String> messages,
                            @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
                            @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics) {
        log.info("开始持久化danmu");
        long start = System.currentTimeMillis();
        if (messages != null && messages.size() > 0) {
            List<BiliUserDanmu> dms = new ArrayList<>();
            List<Long> videoIds = new ArrayList<>();
            //记录弹幕id和videoId
            messages.parallelStream().forEach(msg -> {
                try {
                    BiliUserDanmu danmu = JSON.parseObject(msg, BiliUserDanmu.class);
                    dms.add(danmu);
                    videoIds.add(danmu.getVideoId());
                } catch (Exception e) {
                    log.info("弹幕ids转换失败~{}", e.toString());
                }
            });
            //根据ids批量处理
            if (!dms.isEmpty()) {
                try {
                    doRdbDanmu(dms);
                } catch (Exception e) {
                    log.info("持久化弹幕失败~{}", e.toString());
                    TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();  //标记为回滚
                }
            }
            if (!videoIds.isEmpty()) {
                try {
                    doRdbVideo(videoIds);
                } catch (Exception e) {
                    log.info("更新视频弹幕失败~{}", e.toString());
                    TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();  //标记为回滚
                }
            }
        }
        long end = System.currentTimeMillis();
        log.info("持久化detail成功,耗时:{}ms", (end - start));
    }

    private void doRdbVideo(List<Long> videoIds) {
        ResponseResult<?> result = videoCrudClient.updateDanmuByVideoIds(videoIds);
        if (!result.getCode().equals(AppHttpCodeEnum.SUCCESS.getCode())) {
            throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
        }
    }

    private void doRdbDanmu(List<BiliUserDanmu> dms) {
        if (biliUserDanmuMapper.insertDanmuByIds(dms) < 1) {
            throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
        }
    }
}
